package com.ffcs.crmd.tsp.task.threadpool;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;

import com.ffcs.crmd.tsp.api.dto.job.TspJobDetailShell;
import com.ffcs.crmd.tsp.api.dto.task.PriorityWeight;
import com.ffcs.crmd.tsp.common.constant.LoggerName;
import com.ffcs.crmd.tsp.common.exception.TspException;
import com.ffcs.crmd.tsp.task.log.TaskLogger;
import com.ffcs.crmd.tsp.task.threadpool.api.IJobProcessRunnable;
import com.ffcs.crmd.tsp.task.threadpool.api.IThreadPool;

/**
 * 
 * 功能说明:任务处理线程池
 *
 * @author ZHONGFUHUA
 * 
 * @since 2.0.0-SNAPSHOT
 *
 */
public class JobProcessThreadPool implements IThreadPool {
    
    private static final Logger            LOG                 = TaskLogger.getLogger(LoggerName.TaskLoggerName);
    
    /**
     * 线程池大小
     */
    private int                            poolSize            = 1;
    
    /**
     * 减少线程池大小锁
     */
    private ReentrantLock                  reduceLock          = new ReentrantLock();
    
    /**
     * 是否启动
     */
    private AtomicBoolean                  isStart             = new AtomicBoolean(false);
    
    /**
     * 是否关闭
     */
    private AtomicBoolean                  isShutdown          = new AtomicBoolean(true);
    
    /**
     * 是否需要减少线程池信号
     */
    private AtomicBoolean                  reduceSignal        = new AtomicBoolean(false);
    
    /**
     * 优先级权重
     */
    private PriorityWeight                 priorityWeight      = new PriorityWeight();
    
    /**
     * 待处理的任务
     */
    private BlockingQueue<JobRunnablePair> jobDetailQueue      = new LinkedBlockingQueue<JobRunnablePair>();
    
    /**
     * 工作线程
     */
    private BlockingQueue<WorkerThread>    workersQueue        = new LinkedBlockingQueue<WorkerThread>();
    
    /**
     * 按优先级分类运行的次数
     */
    private Map<Integer, AtomicLong>       runningCount        = new HashMap<Integer, AtomicLong>();
    
    /**
     * 按优先级分类待处理的任务
     */
    private Map<Integer, AtomicLong>       jobDetailNumberMap  = new HashMap<Integer, AtomicLong>();
    
    /**
     * 运行中的线程和子任务
     */
    private Map<String, JobRunnablePair>   jobDetailRunningMap = new ConcurrentHashMap<String, JobRunnablePair>();
    
    /**
     * 线程组
     */
    private ThreadGroup                    threadGroup         = new ThreadGroup("TspJobProcessThreadGroup");
    
    public JobProcessThreadPool() {
        
    }
    
    public JobProcessThreadPool(int poolSize) {
        if(poolSize > 0){
            this.poolSize = poolSize;
        }
    }
    
    @Override
    public void start() throws TspException {
        if (isStart.get()) {
            throw new TspException("This thread pool has been started, Please don't repeat start.");
        }
        
        if (poolSize <= 0) {
            throw new TspException("Thread pool size must be > 0");
        }
        
        runningCount.put(new Integer(PriorityWeight.PRIORITY_FIVE), new AtomicLong(0));
        runningCount.put(new Integer(PriorityWeight.PRIORITY_FOUR), new AtomicLong(0));
        runningCount.put(new Integer(PriorityWeight.PRIORITY_THREE), new AtomicLong(0));
        runningCount.put(new Integer(PriorityWeight.PRIORITY_TWO), new AtomicLong(0));
        runningCount.put(new Integer(PriorityWeight.PRIORITY_ONE), new AtomicLong(0));
        
        jobDetailNumberMap.put(new Integer(PriorityWeight.PRIORITY_FIVE), new AtomicLong());
        jobDetailNumberMap.put(new Integer(PriorityWeight.PRIORITY_FOUR), new AtomicLong());
        jobDetailNumberMap.put(new Integer(PriorityWeight.PRIORITY_THREE), new AtomicLong());
        jobDetailNumberMap.put(new Integer(PriorityWeight.PRIORITY_TWO), new AtomicLong());
        jobDetailNumberMap.put(new Integer(PriorityWeight.PRIORITY_ONE), new AtomicLong());
        
        isStart.compareAndSet(false, true);
        isShutdown.compareAndSet(true, false);
        createWorkerThreads(poolSize);
    }
    
    @Override
    public void shutdown() throws TspException {
        if (isShutdown.get()) {
            throw new TspException("This thread pool has been shutdowned, Please don't repeat shutdown.");
        }
        isShutdown.compareAndSet(false, true);
        isStart.compareAndSet(true, false);
        for (WorkerThread workerThread : workersQueue) {
            workerThread.shutdown();
        }
        
        while (workersQueue.size() > 0) {
            WorkerThread workerThread = workersQueue.poll();
            try {
                if (workerThread != null && workerThread.isRunning()) {
                    LOG.debug("TASK-THREAD-POOL: waiting for thread {} to shutdown", workerThread.getName());
                    Thread.sleep(2000);
                    workersQueue.put(workerThread);
                }
            } catch (InterruptedException e) {
                LOG.error("TASK-THREAD-POOL: shutdown thread pool is interrupted exception:", e);
            }
        }
        LOG.info("TASK-THREAD-POOL: shut down thread pool success.");
    }
    
    @Override
    public int getPoolSize() {
        return poolSize;
    }
    
    @Override
    public void updatePoolSize(int poolSize) {
        LOG.debug("TASK-THREAD-POOL: update thread pool size[{}] to {}.", this.poolSize, poolSize);
        if (this.poolSize > poolSize) {
            reduceSignal.set(true);
        } else if (this.poolSize < poolSize) {
            createWorkerThreads(poolSize - this.poolSize);
        }
        this.poolSize = poolSize;
    }
    
    @Override
    public void updatePriority(PriorityWeight priorityWeight) {
        this.priorityWeight = priorityWeight;
    }
    
    /**
     * 
     * 功能说明:提交子任务
     * 
     * @param tspJobDetialShell 子任务运行内核
     * @param runnable 线程
     * @throws TspException 
     */
    public void submit(TspJobDetailShell tspJobDetialShell, int priority, IJobProcessRunnable runnable) {
        JobRunnablePair jobRunnablePair = new JobRunnablePair(tspJobDetialShell, runnable);
        if (priority < 1 || priority > 5) {
            priority = 1;
        }
        AtomicLong jobDetailNumber = jobDetailNumberMap.get(priority);
        jobDetailNumber.incrementAndGet();
        jobDetailQueue.add(jobRunnablePair);
    }
    
    /**
     * 
     * 功能说明:获取线程
     * 
     * @param id  子任务ID
     * @return 运行线程
     */
    public JobRunnablePair getThread(String id) {
        return jobDetailRunningMap.get(id);
    }
    
    /**
     * 功能说明: 打印线程池信息
     * 
     */
    public void printInfo() {
        LOG.debug("------------------------------------------------------------------------------------------");
        LOG.debug("|                                  JobProcessThreadPool                                  |");
        LOG.debug("------------------------------------------------------------------------------------------");
        LOG.debug("|  pool size    | {}                                                                    |", this.poolSize);
        LOG.debug("|  status       | {}                                                              |", this.isStart.get() ? "RUNNING"
            : "SHUTDOWN");
        LOG.debug("------------------------------------------------------------------------------------------");
        LOG.debug("|  priority[5]  | waiting : {} - running : {}                                         |",
            jobDetailNumberMap.get(PriorityWeight.PRIORITY_FIVE).get(), runningCount.get(PriorityWeight.PRIORITY_FIVE));
        LOG.debug("|  priority[4]  | waiting : {} - running : {}                                         |",
            jobDetailNumberMap.get(PriorityWeight.PRIORITY_FOUR).get(), runningCount.get(PriorityWeight.PRIORITY_FOUR));
        LOG.debug("|  priority[3]  | waiting : {} - running : {}                                         |",
            jobDetailNumberMap.get(PriorityWeight.PRIORITY_THREE).get(), runningCount.get(PriorityWeight.PRIORITY_THREE));
        LOG.debug("|  priority[2]  | waiting : {} - running : {}                                         |",
            jobDetailNumberMap.get(PriorityWeight.PRIORITY_TWO).get(), runningCount.get(PriorityWeight.PRIORITY_TWO));
        LOG.debug("|  priority[1]  | waiting : {} - running : {}                                         |",
            jobDetailNumberMap.get(PriorityWeight.PRIORITY_ONE).get(), runningCount.get(PriorityWeight.PRIORITY_ONE));
        LOG.debug("------------------------------------------------------------------------------------------");
    }
    
    /**
     * 
     * 功能说明:创建线程
     * 
     * @param createCount 创建数量
     */
    private void createWorkerThreads(int createCount) {
        for (int i = 1; i <= createCount; ++i) {
            WorkerThread wt = new WorkerThread(threadGroup, "TspJobProcessThread-" + i);
            wt.start();
            workersQueue.add(wt);
        }
    }
    
    /**
     * 
     * 功能说明:工作线程
     *
     * @author ZHONGFUHUA
     * 
     * @since 2.0.0-SNAPSHOT
     *
     */
    class WorkerThread extends Thread {
        
        private AtomicBoolean shutdown = new AtomicBoolean(true);
        
        private AtomicBoolean running  = new AtomicBoolean(false);
        
        WorkerThread(ThreadGroup threadGroup, String name) {
            super(threadGroup, name);
        }
        
        public void shutdown() {
            shutdown.set(true);
            super.interrupt();
            LOG.info("TASK-THREAD-POOL: notify worker thread to shut down.");
        }
        
        @Override
        public void run() {
            while (!shutdown.get()) {
                // 判断是否有信号通知要减少线程池，如果有，则进行关闭该线程操作
                if (reduceSignal.get()) {
                    reduceLock.lock();
                    try {
                        if (reduceSignal.get()) {
                            shutdown.compareAndSet(false, true);
                        }
                        if (workersQueue.size() > poolSize) {
                            workersQueue.remove(this);
                            if (workersQueue.size() == poolSize) {
                                reduceSignal.compareAndSet(true, false);
                            }
                            LOG.debug("TASK-THREAD-POOL: worker thread will shut down because thread pool reduce pool size.");
                        }
                    } finally {
                        reduceLock.unlock();
                    }
                }
                
                if (shutdown.get()) {
                    break;
                }
                
                try {
                    JobRunnablePair jobRunnablePair = jobDetailQueue.take();
                    boolean runFlag = false;
                    int priority = jobRunnablePair.getTspJobDetailShell().getPriority();
                    
                    if (isNonExistOther(priority)) {
                        runFlag = true;
                    } else if (isNonOverWorkers()) {
                        runFlag = true;
                    } else {
                        float percent = getPercent(priority);
                        float weight = getWeight(priority);
                        if (weight <= 0) {
                            LOG.warn("TASK-THREAD-POOL: can't to run a runnable because weight[{}] <= 0", weight);
                        } else if (percent <= weight) {
                            // 如果当前优先级运行次数百分 小于设定的权重，则执行当前任务
                            runFlag = true;
                        }
                    }
                    
                    if (runFlag) {
                        String id = jobRunnablePair.getTspJobDetailShell().getId();
                        jobRunnablePair.setThread(this);
                        Runnable runnable = jobRunnablePair.getRunnable();
                        runningCount.get(priority).incrementAndGet();
                        jobDetailRunningMap.put(id, jobRunnablePair);
                        try {
                            runnable.run();
                        } catch (Throwable e) {
                            LOG.debug("TASK-THREAD-POOL: run a job process business data runnable exception:", e);
                        }
                        jobDetailRunningMap.remove(id);
                        jobDetailNumberMap.get(priority).decrementAndGet();
                        runningCount.get(priority).decrementAndGet();
                    } else {
                        jobDetailQueue.put(jobRunnablePair);
                    }
                } catch (InterruptedException e) {
                    LOG.debug("TASK-THREAD-POOL: take a job runnable is interrupted.");
                }
            }
            running.compareAndSet(true, false);
            LOG.info("TASK-THREAD-POOL: worker thread is shut down.");
        }
        
        /**
         * 
         * 功能说明:获取对应优先级的运行数量占总量的百分比
         * 
         * @param priority 优先级
         * @return 已经运行的数量占总量的百分比
         */
        private float getPercent(int priority) {
            int totalCount = 0;
            for (AtomicLong atomicLong : runningCount.values()) {
                totalCount += atomicLong.get();
            }
            
            if (totalCount == 0) {
                return 0;
            } else {
                return Float.valueOf(runningCount.get(priority).get()) / Float.valueOf(totalCount);
            }
        }
        
        /**
         * 
         * 功能说明:不存在其他优先级的任务
         * 
         * @param priority
         * @return
         */
        private boolean isNonExistOther(int priority) {
            for (int i = 1; i <= 5; i++) {
                if (i != priority) {
                    if (jobDetailNumberMap.get(i).get() != 0) {
                        return false;
                    }
                }
            }
            return true;
        }
        
        /**
         * 
         * 功能说明:待执行任务是否超过工作线程
         * 
         * @return
         */
        private boolean isNonOverWorkers() {
            return workersQueue.size() > jobDetailQueue.size() ? true
                : false;
        }
        
        /**
         * 
         * 功能说明:获取权重
         * 
         * @param priority 优先级
         * @return 权重
         */
        private float getWeight(int priority) {
            return priorityWeight.getPriorityWeight(priority);
        }
        
        /**
         * 功能说明:启动工作线程
         * 
         */
        @Override
        public synchronized void start() {
            super.start();
            shutdown.compareAndSet(true, false);
            running.compareAndSet(false, true);
        }
        
        /**
         * 功能说明:判断是否运行中
         *
         * @return 是否运行中
         */
        public boolean isRunning() {
            return running.get();
        }
    }
    
}
